package net.bwie.realtime.jtp.dwd.logs.car.job;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.jtp.utils.DorisUtil;
import net.bwie.realtime.jtp.utils.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.LocalStreamEnvironment;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.List;
import java.util.Map;

public class MotorHighTemperatureCEPJob {
    public static void main(String[] args) throws Exception {
        // env执行环境
        LocalStreamEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(3000L);
        // 消费kafka
        DataStream<String> kafkaStream = KafkaUtil.consumerKafka(env, "dwd-car-logs");
//        kafkaStream.print("kafka");
        //转实体类
        SingleOutputStreamOperator<CarBean> beanStream = kafkaStream.map(new MapFunction<String, CarBean>() {
            @Override
            public CarBean map(String s) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);
                String vin = jsonObject.getString("vin");
                Long timestamp = jsonObject.getLong("timestamp");
                int type = jsonObject.getIntValue("type");
                double longitude = jsonObject.getDoubleValue("longitude");
                double latitude = jsonObject.getDoubleValue("latitude");
                double speed = jsonObject.getDoubleValue("speed");
                double mileage = jsonObject.getDoubleValue("mileage");
                double batteryLevel = jsonObject.getDoubleValue("batteryLevel");
                double batteryTemp = jsonObject.getDoubleValue("batteryTemp");
                double motorTemp = jsonObject.getDoubleValue("motorTemp");
                String chargingStatus = jsonObject.getString("chargingStatus");
                double energyConsumption = jsonObject.getDoubleValue("energyConsumption");
                return new CarBean(
                        vin,
                        timestamp,
                        type,
                        longitude,
                        latitude,
                        speed,
                        mileage,
                        batteryLevel,
                        batteryTemp,
                        motorTemp,
                        chargingStatus,
                        energyConsumption
                );
            }
        });
//        beanStream.print();
        //添加水位线
        SingleOutputStreamOperator<CarBean> timeStream = beanStream.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<CarBean>forBoundedOutOfOrderness(Duration.ofSeconds(1))
                        .withTimestampAssigner(new SerializableTimestampAssigner<CarBean>() {
                            @Override
                            public long extractTimestamp(CarBean carBean, long l) {
                                return carBean.getTimestamp();
                            }
                        })
        );
        //CEP 连续三次低电量
        Pattern<CarBean, ?> pattern = Pattern.<CarBean>begin("one")
                .where(SimpleCondition.of(event -> event.getMotorTemp() < 60.0))
                .next("two")
                .where(SimpleCondition.of(subEvent -> subEvent.getMotorTemp() < 60.0))
                .followedBy("three")
                .where(SimpleCondition.of(event -> event.getMotorTemp() < 60.0));

        PatternStream<CarBean> patternStream = CEP.pattern(timeStream.keyBy(a -> a.getVin()), pattern);
        DataStream<String> lowBatteryPower = patternStream.select(new PatternSelectFunction<CarBean, String>() {
            @Override
            public String select(Map<String, List<CarBean>> map) throws Exception {
                String vin = map.get("three").get(0).getVin();
                Long timestamp = map.get("three").get(0).getTimestamp();
                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                String times = sdf.format(timestamp);
                return vin+","+"电机温度过高"+","+times;
            }
        });
        //输出
        lowBatteryPower.print();
        //存入Doris
        DorisUtil.saveToDoris(
                lowBatteryPower,
                "ner_car_report",
                "dws_cars_err"
        );
        // 触发执行
        env.execute("MotorHighTemperatureCEPJob");
    }
}
